package com.zx.learn.flink.sink;

import com.zx.learn.flink.utils.DataUtils;
import com.zx.learn.flink.utils.NcMockServer;
import com.zx.learn.flink.utils.PathUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;

@Slf4j
public class WriteSinkDemo {
    public static void main(String[] args) throws Exception {
        log.info("生成NC数据");
        List<String> ncData = DataUtils.getData("wordcount.txt");
        NcMockServer.generateData(ncData);
        //1）获取flink流处理的运行环境
        log.info("处理数据");
        //local模式默认的并行度是当前机器的逻辑核的数量
        Configuration configuration = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        int parallelism0 = env.getParallelism();
        log.info("执行环境默认的并行度：" + parallelism0);
        DataStreamSource<String> lines = env.socketTextStream("localhost", NcMockServer.PORT);
        //获取DataStream的并行度
        int parallelism = lines.getParallelism();
        log.info("SocketSource的并行度：" + parallelism);
        //StreamingFileSink
        String fullFilePathName = PathUtil.CLASSPATH_DATA_OUTPUT + "parallelism/";
        log.info("输出路径:{}", fullFilePathName);
        lines.writeAsText(fullFilePathName, FileSystem.WriteMode.OVERWRITE).setParallelism(2);
        env.execute();
    }
}
